package com.om.opensourway;

import Utils.HttpClientUtils;
import com.om.common.CommonObject;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.ObjectNode;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import scala.Tuple2;

import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.URISyntaxException;
import java.net.URL;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Comparator.comparing;

public class CombineIsoRpmDownloadData {
    public static void main(String[] args) throws URISyntaxException, ClassNotFoundException, IllegalAccessException, InstantiationException {
        String taskCount = args[2];
        String inputPath = args[3];
        String vhost = args[4];
        String esIndexName = args[5];
        String className = args[7];
        String timeInterval = args[8];

        CommonObject commonObject = (CommonObject) Class.forName(className).newInstance();
        commonObject.setEsIndexName(esIndexName);

        SparkConf sparkConf = commonObject.getSparkConf(args);
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        Broadcast<CommonObject> broCommonObject = jsc.broadcast(commonObject);
        JavaRDD<String> srcRdd = jsc.textFile(inputPath, Integer.parseInt(taskCount));
        JavaRDD<String> filterRdd = filterLog(srcRdd, vhost);
        JavaRDD<ArrayList<String>> combineRdd = dataCombine(filterRdd, timeInterval, jsc);
        pushToEs(combineRdd, broCommonObject);

    }

    private static JavaRDD<String> filterLog(JavaRDD<String> srcRdd, String host) {
        return srcRdd.filter((Function<String, Boolean>) data -> {
            String vhost;
            try {
                String log = CommonObject.objectMapper.readTree(data).get("_source").get("log").asText();
                JsonNode lognode = CommonObject.objectMapper.readTree(log);
                vhost = lognode.get("vhost").getTextValue();
                String path = lognode.get("path").getTextValue();
                if ((StringUtils.isBlank(path) || path.endsWith("/") || path.equals("-"))) return false;
            } catch (Exception e) {
                return false;
            }
            return host.equals(vhost);
        });
    }

    private static JavaRDD<ArrayList<String>> dataCombine(JavaRDD<String> srcRdd, String timeInterval, JavaSparkContext jsc) {
        JavaPairRDD<String, Iterable<String>> pairRdd = srcRdd.mapToPair((PairFunction<String, String, String>) jsonNode1 -> {
            String str = CommonObject.objectMapper.readTree(jsonNode1).get("_source").get("log").asText();
            JsonNode jsonNode = CommonObject.objectMapper.readTree(str);
            System.out.println("jsonNode: " + jsonNode);
            String ip = jsonNode.get("proxy_remote_addr").getTextValue();

            //如果proxy_remote_addr没有数据则取remote_add的ip
            if (StringUtils.isBlank(ip) || "-".equals(ip)) {
                ip = jsonNode.get("remote_addr").getTextValue();
            }
            String path = jsonNode.get("path").getTextValue();
            return new Tuple2<>(ip + path, str);
        }).groupByKey();

        MyAccumulator myAccumulator = new MyAccumulator();
        jsc.sc().register(myAccumulator, "myAccumulator");
        Broadcast<MyAccumulator> myAccumulatorBc = jsc.broadcast(myAccumulator);

        JavaRDD<ArrayList<String>> combineRdd = pairRdd.values().map(it -> {
            ArrayList<JsonNode> jsonNodes = new ArrayList<>();
            Iterator<String> iterator = it.iterator();
            while (iterator.hasNext()) {
                try {
                    jsonNodes.add(CommonObject.objectMapper.readTree(iterator.next()));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            jsonNodes.sort(comparing(o -> o.get("time").getTextValue()));

            DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss+00:00");
            ArrayList<String> res = new ArrayList<>();

            LocalDateTime startTime = null;
            LocalDateTime endTime = null;
            JsonNode rangeStartNode = jsonNodes.stream().filter(jsonNode -> {
                String path = jsonNode.get("path").getTextValue();
                if (path.endsWith(".iso") && jsonNode.get("status").getIntValue() == 206) {
                    String httpRange = jsonNode.get("http_range").getTextValue();
                    return StringUtils.isNotBlank(httpRange) && httpRange.startsWith("bytes=0-");
                } else return false;
            }).findFirst().orElse(null);
            if (rangeStartNode != null) startTime = LocalDateTime.parse(rangeStartNode.get("time").getTextValue(), dtf);

            List<JsonNode> rangeEndNodes = jsonNodes.stream().filter(jsonNode -> {
                String path = jsonNode.get("path").getTextValue();
                if (path.endsWith(".iso") && jsonNode.get("status").getIntValue() == 206) {
                    long fileActualSize = getFileActualSize("https://repo.openeuler.org" + path, myAccumulatorBc.value().value());
                    String httpRange = jsonNode.get("http_range").getTextValue();
                    return StringUtils.isNotBlank(httpRange) && httpRange.endsWith("-" + (fileActualSize - 1));
                } else return false;
            }).collect(Collectors.toList());
            if (!rangeEndNodes.isEmpty()) {
                endTime = LocalDateTime.parse(rangeEndNodes.get(rangeEndNodes.size() - 1).get("time").getTextValue(), dtf);
                res.addAll(rangeEndNodes.stream().map(JsonNode::toString).collect(Collectors.toList()));
            }

            ArrayList<JsonNode> filterJsonNodes = new ArrayList<>();
            if (startTime != null && endTime == null) {
                LocalDateTime finalStartTime = startTime;
                List<JsonNode> beforeStartTimeNodes = jsonNodes.stream().filter(jsonNode -> {
                    LocalDateTime time = LocalDateTime.parse(jsonNode.get("time").getTextValue(), dtf);
                    return time.isBefore(finalStartTime.minusMinutes(Long.parseLong(timeInterval))) ||
                            !jsonNode.get("path").getTextValue().endsWith(".iso") ||
                            jsonNode.get("status").getIntValue() != 206;
                }).collect(Collectors.toList());
                filterJsonNodes.addAll(beforeStartTimeNodes);
                res.add(rangeStartNode.toString());
            } else if (endTime != null && startTime == null) {
                LocalDateTime finalEndTime = endTime;
                List<JsonNode> endTimeAfterNodes = jsonNodes.stream().filter(jsonNode -> {
                    LocalDateTime time = LocalDateTime.parse(jsonNode.get("time").getTextValue(), dtf);
                    return time.isAfter(finalEndTime.plusMinutes(Long.parseLong(timeInterval))) ||
                            !jsonNode.get("path").getTextValue().endsWith(".iso") ||
                            jsonNode.get("status").getIntValue() != 206;
                }).collect(Collectors.toList());
                filterJsonNodes.addAll(endTimeAfterNodes);
            } else if (startTime != null && endTime != null) {
                LocalDateTime finalEndTime1 = endTime;
                LocalDateTime finalStartTime1 = startTime;
                List<JsonNode> beforeStartAndAfterEndNodes = jsonNodes.stream().filter(jsonNode -> {
                    LocalDateTime time = LocalDateTime.parse(jsonNode.get("time").getTextValue(), dtf);
                    return time.isAfter(finalEndTime1.plusMinutes(Long.parseLong(timeInterval))) ||
                            time.isBefore(finalStartTime1.minusMinutes(Long.parseLong(timeInterval))) ||
                            !jsonNode.get("path").getTextValue().endsWith(".iso") ||
                            jsonNode.get("status").getIntValue() != 206;
                }).collect(Collectors.toList());
                filterJsonNodes.addAll(beforeStartAndAfterEndNodes);
            } else {
                filterJsonNodes.addAll(jsonNodes);
            }

            long sumBytesSent = 0L;
            JsonNode lastNode = null;
            JsonNode finalNode = null;
            for (JsonNode jsonNode : filterJsonNodes) {
                if (jsonNode.get("path").getTextValue().endsWith(".iso") &&
                        jsonNode.get("status").getIntValue() == 206) {
                    finalNode = jsonNode;
                    long bytesSent = jsonNode.get("bytes_sent").getLongValue();
                    sumBytesSent += bytesSent;

                    LocalDateTime time = LocalDateTime.parse(jsonNode.get("time").getTextValue(), dtf);
                    if (lastNode == null) lastNode = jsonNode;
                    Duration between = Duration.between(LocalDateTime.parse(lastNode.get("time").getTextValue(), dtf), time);
                    if (between.toMinutes() > Long.parseLong(timeInterval)) {
                        ObjectNode objectNode = (ObjectNode) lastNode;
                        objectNode.put("bytes_sent", sumBytesSent - bytesSent);
                        res.add(objectNode.toString());
                        sumBytesSent = bytesSent;
                    }
                    lastNode = jsonNode;
                } else {
                    res.add(jsonNode.toString());
                }
            }

            if (finalNode != null) {
                ObjectNode objectNode = (ObjectNode) finalNode;
                objectNode.put("bytes_sent", sumBytesSent);
                res.add(objectNode.toString());
            }

            return res;
        });

        return combineRdd;
    }

    private static void pushToEs(JavaRDD<ArrayList<String>> combineRdd, Broadcast<CommonObject> broCommonObject) {
        combineRdd.foreachPartition((VoidFunction<Iterator<ArrayList<String>>>) it -> {
            CloseableHttpClient client = HttpClientUtils.getClient();
            RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
            BulkRequest request = new BulkRequest();

            while (it.hasNext()) {
                for (String s : it.next()) {
                    JsonNode log = CommonObject.objectMapper.readTree(s);
                    HashMap<String, String> resMap = broCommonObject.value().assemblydataNew(client, log, null);
                    String id = DigestUtils.md5Hex(resMap.get("created_at") + resMap.get("ip") + resMap.get("path"));
                    request.add(new IndexRequest(broCommonObject.value().getEsIndexName(), "_doc", id).source(resMap));
                }
            }
            if (request.requests().size() != 0) restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            restHighLevelClient.close();
        });
    }

    private static long getFileActualSize(String path, HashMap<String, Long> path2size) {
        String pathLower = path.toLowerCase();
        if (path2size.containsKey(pathLower)) return path2size.get(pathLower);

        long size = -1;
        HttpURLConnection conn = null;
        try {
            URL url = new URL(path);
            conn = (HttpURLConnection) url.openConnection();
            size = conn.getContentLengthLong();
            if (size < 0 || size == 169) size = -1;
            path2size.put(pathLower, size);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            assert conn != null;
            conn.disconnect();
        }

        return size;
    }
}
